package com.zz.common.zid

import java.time.Instant

import cats.data.OptionT
import cats.effect.IO
import cats.implicits._
import com.aerospike.client.{Bin, Record}
import com.zz.ady.idl.IdType
import com.zz.aerospike.AsDAO
import com.zz.common.service.IdService.Zid
import com.zz.idl.{IdBundle, IdBundleMeta, ReversedIndexMeta, Zids}

class AerospikeZidStorage(asDao:AsDAO[IO],
                          indexNamespace:String,
                          indexSet:String,
                          reversedIndexNamespace:String) extends ZidStorage[IO] {
  override def storeIdBundle(idBundle: IdBundle, idBundleMeta: IdBundleMeta): IO[Unit] = {
    val idBundleBin = new Bin("ids", idBundle.toByteArray)
    val metaBin = new Bin("meta", idBundleMeta.toByteArray)
    asDao.as.put(indexNamespace, indexSet, idBundle.zid, Seq(idBundleBin, metaBin))
  }

  override def storeReversedIndex(id: String, idType: IdType, zid: String, reversedIndexMeta: ReversedIndexMeta): IO[Unit] = {
    asDao.as.get(reversedIndexNamespace, idType.toString(), id) >>= {
      case Some(record)=>
        val zids = Zids.parseFrom(record.getValue("ids").asInstanceOf[Array[Byte]])
        val newZids = Set.from(zids.zids :+ zid)
        val zidBin = new Bin("id", zid)
        val zidsBin = new Bin("ids", Zids(newZids.toVector).toByteArray)
        val metaBin = new Bin("meta", new ReversedIndexMeta(Instant.now.toEpochMilli).toByteArray)
        asDao.as.put(reversedIndexNamespace, idType.toString, id, Seq(zidBin, zidsBin, metaBin))
      case None=>
        val zidBin = new Bin("id", zid)
        val zidsBin = new Bin("ids",Zids(Vector(zid)).toByteArray)
        val metaBin = new Bin("meta", ReversedIndexMeta(Instant.now().toEpochMilli).toByteArray)
        asDao.as.put(reversedIndexNamespace, idType.toString, id, Seq(zidBin, zidsBin, metaBin))
    }
  }

  override def queryZidInfo(zid: String): OptionT[IO, (IdBundle, IdBundleMeta)] =
    for{
      record <- OptionT(asDao.as.get(indexNamespace,indexSet,zid))
      idBundle <- OptionT.liftF(IO(IdBundle.parseFrom(record.getValue("ids").asInstanceOf[Array[Byte]])))
      meta <- OptionT.liftF(IO(IdBundleMeta.parseFrom(record.getValue("meta").asInstanceOf[Array[Byte]])))
    }yield (idBundle, meta)

  override def queryZids(id: String, idType: IdType): OptionT[IO, (Zid, Seq[Zid], ReversedIndexMeta)] =
    for{
      record <- OptionT(asDao.as.get(reversedIndexNamespace, idType.toString(), id))
      zid = record.getString("id")
      zids = Zids.parseFrom(record.getValue("ids").asInstanceOf[Array[Byte]]).zids
      meta = ReversedIndexMeta.parseFrom(record.getValue("meta").asInstanceOf[Array[Byte]])
    }yield(zid, zids, meta)
}
